package com.topnav.mc.sse;

import com.alibaba.fastjson2.JSONObject;
import com.js.lib.cache.SysCache;
import com.js.lib.result.Result;
import com.js.lib.utils.DateUtils;
import com.js.lib.utils.JsonUtil;
import com.topnav.mc.admin.entity.DeviceAlarm;
import com.topnav.mc.admin.entity.bean.DeviceAlarmBean;
import com.topnav.mc.admin.security.PermissionGrade;
import com.topnav.mc.common.Constant;
import com.topnav.mc.lock.SyncLock;
import com.topnav.mc.uitls.PermissionGradeUtil;
import com.topnav.mc.uitls.TaskLockUtil;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 设备报警推送通知队列
 */
@Slf4j
@Component
public class DeviceAlarmQueue {
    private LinkedBlockingDeque<MessageBean> deque = null;

    @Autowired
    private SysCache sysCache;
    private TaskLockUtil taskLockUtil;
    @Resource
    PermissionGradeUtil permissionGradeUtil;

    @PostConstruct
    public void DeviceAlarmQueue(){
        deque = new  LinkedBlockingDeque(200);
    }

    /**
     * 取得堆积数量
     * @return
     */
    public Integer getQueueCount(){
        return deque.size();
    }

    /**
     * Offer
     * @param data
     */
    public void offer(MessageBean data){
        deque.offer(data);
        this.send();
    }

    /**
     * 出现错误： ResponseBodyEmitter is already set complete
     * 是因为多线程 线程安全问题
     * 参阅：https://www.5axxw.com/questions/simple/jzlft5
     */
    @SyncLock(key="DeviceAlarmQueue-send-LOCK",dueSec = 5*60,errorUnLock = true)
    @Async("taskExecutor")
    public void send() {
        while(!this.deque.isEmpty()){
            MessageBean message = this.deque.poll();
            Map<Object,Object> emitterMap =  sysCache.hmget(Constants.CAHCE_SSE_CLIENT);
            if (emitterMap == null){
                continue;
            }

           if (message.getType() == EnumSseMessageType.ALARM_NOTIFY){
                //推送报警信息
               this.sseSendAlarm(emitterMap,message);
           }else if (message.getType() == EnumSseMessageType.UP_DOWN_LINE_NOTIFY){
               //设备上下线通知
               this.sseSendUpDownLine(emitterMap,message);
           }else if (message.getType() == EnumSseMessageType.MEDIA_REPLAY_FINISHED){
               //录像回放结束通知
               this.sseSendReplayFinish(emitterMap,message);
           }
        }
    }

    /**
     * 报警消息推送
     * @param emitterMap
     * @param message
     */
    private void sseSendAlarm(Map<Object,Object> emitterMap,MessageBean message){
        PermissionGrade permissionGrade = null;
        try{
            DeviceAlarmBean deviceAlarmBean = (DeviceAlarmBean)message.getMessage();
            String msg = deviceAlarmBean.getAlarmMessage();
            if (StringUtils.isEmpty(msg)){
                msg = String.format("%s %s %s 的设备(%s)在(%f,%f)位置 %s 报警",
                        DateUtils.format(deviceAlarmBean.getAlarmTime(),"yyyy-MM-dd HH:mm:ss"),
                        deviceAlarmBean.getShortName(),
                        deviceAlarmBean.getUserName(),
                        deviceAlarmBean.getDeviceCode(),
                        deviceAlarmBean.getLongitude(),
                        deviceAlarmBean.getLatitude(),
                        deviceAlarmBean.getAlarmCateDesc());
            }
            //message.setMessage(msg);
            SseMessage sseMessage = new SseMessage(EnumSseMessageType.ALARM_NOTIFY,deviceAlarmBean.getId(),deviceAlarmBean.getAlarmCateDesc(),deviceAlarmBean,msg);
            sseMessage.getData().put("time",DateUtils.format(deviceAlarmBean.getAlarmTime(),"yyyy-MM-dd HH:mm:ss"));
            sseMessage.getData().put("organization",deviceAlarmBean.getShortName());
            sseMessage.getData().put("user",deviceAlarmBean.getUserName());
            sseMessage.getData().put("device",deviceAlarmBean.getDeviceCode());
            sseMessage.getData().put("position",deviceAlarmBean.getPosition());
            sseMessage.getData().put("alarmCate",deviceAlarmBean.getAlarmCate().toString());
            sseMessage.setObject(null);

            this.sseSend(emitterMap,sseMessage,message);
            /*for (Iterator<Map.Entry<Object, Object>> it = emitterMap.entrySet().iterator(); it.hasNext();) {
                Map.Entry<Object,Object> item = it.next();
                try{
                    EmitterBean emitterBean = (EmitterBean)item.getValue();
                    //权限判断
                    permissionGrade = permissionGradeUtil.getPmGrade(emitterBean.getToken());
                    if (permissionGrade == null || !permissionGrade.getEnabledNotify()){
                        //不允许接收通知
                        continue;
                    }
                    //分权限
                    if (permissionGrade.getPermissionLevel() == 1){
                        if (permissionGrade.getUserId().equals(message.getUserId())){
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() == 2){
                        if (permissionGrade.getOrgId().equals(message.getOrgId())){
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() == 3 || permissionGrade.getPermissionLevel() == 4 ) {
                        if (message.getOrgPath().startsWith(permissionGrade.getOrgPath())) {
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else{
                        emitterBean.getEmitter().send(Result.ok(sseMessage));
                    }
                }catch (Exception e){
                    log.error(">>>>>>> 消息推送给【{}】失败,id={} ** {}",permissionGrade.getUserName(),item.getKey().toString(),e.toString());
                    sysCache.hdel(Constants.CAHCE_SSE_CLIENT,item.getKey().toString());
                }

            }*/
        }catch (Exception e){
            log.error(">>>>>>> 消息推送给 【{}】 失败: {}",permissionGrade.getUserName(),e.toString());
            //taskLockUtil.unLock("DeviceAlarmQueue-send-LOCK");
        }

    }

    /**
     * 上下线消息推送
     * @param emitterMap
     * @param message
     */
    private void sseSendUpDownLine(Map<Object,Object> emitterMap,MessageBean message){
        PermissionGrade permissionGrade = null;
        try{

            Map<String,Object> map  = (Map)message.getMessage();

/*            String msg = deviceAlarmBean.getAlarmMessage();
            if (StringUtils.isEmpty(msg)){
                msg = String.format("%s %s %s 的设备(%s)在(%f,%f)位置 %s 报警",
                        DateUtils.format(deviceAlarmBean.getAlarmTime(),"yyyy-MM-dd HH:mm:ss"),
                        deviceAlarmBean.getShortName(),
                        deviceAlarmBean.getUserName(),
                        deviceAlarmBean.getDeviceCode(),
                        deviceAlarmBean.getLongitude(),
                        deviceAlarmBean.getLatitude(),
                        deviceAlarmBean.getAlarmCateDesc());
            }*/

            //message.setMessage(msg);
            SseMessage sseMessage = new SseMessage(
                    EnumSseMessageType.UP_DOWN_LINE_NOTIFY,
                    Long.valueOf(map.get("id").toString()),
                    "0".equals(map.get("action").toString())?"offline":"online",
                    map,
                    null);
            this.sseSend(emitterMap,sseMessage,message);

            /*for (Iterator<Map.Entry<Object, Object>> it = emitterMap.entrySet().iterator(); it.hasNext();) {
                Map.Entry<Object,Object> item = it.next();
                try{
                    EmitterBean emitterBean = (EmitterBean)item.getValue();
                    if (emitterBean == null){
                        continue;
                    }
                    //权限判断
                    permissionGrade = permissionGradeUtil.getPmGrade(emitterBean.getToken());
                    if (permissionGrade == null){
                        log.error(">>>>>> token: {} 登录已失效。",emitterBean.getToken());
                        continue;
                    }
                    //分权限
                    if (permissionGrade.getPermissionLevel() == 1){
                        if (permissionGrade.getUserId().equals(message.getUserId())){
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() == 2){
                        if (permissionGrade.getOrgId().equals(message.getOrgId())){
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() == 3 || permissionGrade.getPermissionLevel() == 4 ) {
                        if (message.getOrgPath().startsWith(permissionGrade.getOrgPath())) {
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() > 4){

                        emitterBean.getEmitter().send(Result.ok(sseMessage));

                    }
                }catch (Exception e){
                    log.error(">>>>>>> 设备上下线推送给 【{}】 id={} 失败: {}",permissionGrade.getUserName(),item.getKey().toString(),e.toString());
                    sysCache.hdel(Constants.CAHCE_SSE_CLIENT,item.getKey().toString());
                }

            }*/
        }catch (Exception e){
            log.error(">>>>>>> 设备上下线推送给 【{}】 失败: {}",permissionGrade.getUserName(),e.toString());
            //taskLockUtil.unLock("DeviceAlarmQueue-send-LOCK");
        }

    }

    /**
     * 回放推流结束推送
     * @param emitterMap
     * @param message
     */
    private void sseSendReplayFinish(Map<Object,Object> emitterMap,MessageBean message){
        PermissionGrade permissionGrade = null;
        try{
            JSONObject json = (JSONObject)message.getMessage();
            SseMessage sseMessage = new SseMessage();
            sseMessage.setType(EnumSseMessageType.MEDIA_REPLAY_FINISHED);
            sseMessage.setObject(json);//内容：{deviceCode:null,finished:false}

            this.sseSend(emitterMap,sseMessage,message);

/*            for (Iterator<Map.Entry<Object, Object>> it = emitterMap.entrySet().iterator(); it.hasNext();) {
                Map.Entry<Object,Object> item = it.next();

                try{
                    EmitterBean emitterBean = (EmitterBean)item.getValue();
                    emitterBean.getEmitter().send(Result.ok(sseMessage));
                }catch (Exception e){
                    log.error(">>>>>>> 设备回放关流通知推送给 【{}】 失败: {}",item.getKey().toString(),e.toString());
                    sysCache.hdel(Constants.CAHCE_SSE_CLIENT,item.getKey().toString());
                }
            }*/

        }catch (Exception e){
            log.error(">>>>>>> 回放关流通知推送给 【{}】 失败: {}",permissionGrade.getUserName(),e.toString());
        }
    }

    /**
     * 给所有在线的页面发送 SSE信息
     * @param emitterMap
     * @param sseMessage
     */
    private void sseSend(Map<Object,Object> emitterMap,SseMessage sseMessage,MessageBean message){
        PermissionGrade permissionGrade = null;
        for (Iterator<Map.Entry<Object, Object>> it = emitterMap.entrySet().iterator(); it.hasNext();) {
            Map.Entry<Object,Object> item = it.next();
            try{


                EmitterBean emitterBean = (EmitterBean)item.getValue();
                if (emitterBean == null){
                    continue;
                }
                if (StringUtils.isEmpty(emitterBean.getToken())){
                    log.error(">>>>>> Empty token sse abort");
                    continue;
                }

                //权限判断
                permissionGrade = permissionGradeUtil.getPmGrade(emitterBean.getToken());
                if (permissionGrade == null){
                    log.error(">>>>>> 登录已失效, 移除 sse");
                    emitterBean.getEmitter().complete();// .send(Result.ok(sseMessage));
                    sysCache.hdel(Constants.CAHCE_SSE_CLIENT,item.getKey().toString());
                    continue;
                }

                if (message.getUserId() != null && message.getOrgId() != null){
                    //分权限
                    if (permissionGrade.getPermissionLevel() == 1){
                        if (permissionGrade.getUserId().equals(message.getUserId())){
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() == 2){
                        if (permissionGrade.getOrgId().equals(message.getOrgId())){
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() == 3 || permissionGrade.getPermissionLevel() == 4 ) {
                        if (message.getOrgPath().startsWith(permissionGrade.getOrgPath())) {
                            emitterBean.getEmitter().send(Result.ok(sseMessage));
                        }
                    }else if (permissionGrade.getPermissionLevel() > 4){

                        emitterBean.getEmitter().send(Result.ok(sseMessage));

                    }
                }else{
                    //不分权限
                    emitterBean.getEmitter().send(Result.ok(sseMessage));
                }

            }catch (Exception e){
                log.error(">>>>>>> {} 异常:{}; 数据：{}",message.getType().getDesc(),e.toString(),sseMessage.toString());
                sysCache.hdel(Constants.CAHCE_SSE_CLIENT,item.getKey().toString());
            }

        }
    }

    public void testSseMessage(){
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("key",123);
        jsonObject.put("name","name");
        SseMessage sseMessage = new SseMessage();
        sseMessage.setType(EnumSseMessageType.ALARM_NOTIFY);
        sseMessage.setDesc("报警通知");
        sseMessage.setMessage("jjkkfdfd");
        sseMessage.setObject(jsonObject);
        System.out.println(sseMessage.toString());
    }

    @Data
    @Accessors(chain=true)
    public class SseMessage{
        private EnumSseMessageType type;
        private Long id;  //数据ID
        private String desc; //数据描述
        private Object object;
        private String message;
        private Map<String,String> data=new HashMap<>();
        public SseMessage(){
            super();
        }
        public SseMessage(EnumSseMessageType type,Long id,String desc,Object object,String message){
            this.type = type;
            this.id = id;
            this.desc = desc;
            this.object = object;
            this.message = message;
        }

        public String toString(){
            return String.format(JsonUtil.toJson(this));
        }
    }
}
